feat: parallel BWAG for the range-window shape (h2o Q8) - #2223

PerPartitionFilterExec's only caller (BallistaAdapter above ShuffleReader
for hash-agg correctness) was already using it as a range-shaped filter.
Widening it into a general per-partition arbitrary-predicate op — with
halo-widening bolted on for the parallel-window rewrite — would leak a
range concept into an arbitrary-predicate contract.

RangeFilterExec is the honest shape: routing_expr + cuts + halo_lo /
halo_hi. Per-partition semantics fall out of the local partition index,
not from a Vec of independent predicates. Ordering knowledge on the
input opens the door to a future ValueIndexReader-driven binary-search
path (PR apache#2204 direction) that a generic FilterExec can't take.

Notable pieces:

- `cuts: Arc<Mutex<Option<Vec<f64>>>>` + `resolve_cuts` API mirror
  `ExchangeExec::range_repartition_routing()` — the ParallelWindow rule
  plants a pending RangeFilterExec at plan time; the scheduler resolves
  cuts after stage 0's RuntimeStatsExec reports merge. `execute` and
  serialization both refuse while cuts are unresolved.
- `partition_indices: Vec<usize>` maps local → global partition index.
  Restrict slices this mapping without touching cuts (cuts stay whole;
  they describe the K global partitions). Replaces PPFE's per-partition
  predicate-vec slicing in task_builder's restrict path.
- Public API + proto speak `ScalarValue` (not `f64`) per the type-
  generality rule for the range-repartition family: the outer contract
  is type-agnostic so KLL can widen internal storage later without an
  API break. Internal downcast to `f64` today; non-Float64 inputs error
  with a clear message.
- Adapter builds RangeFilterExec with `halo=0` for the existing hash-agg
  case; the parallel-window rule will build it with non-zero halo.

Migration: delete `PerPartitionFilterExec`, migrate all callers, rename
proto `PerPartitionFilterExecNode` → `RangeFilterExecNode`, update doc
comments. Full test suite (597 tests) passes.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

feat(scheduler): ParallelWindowRule — distributed range-shuffle for BWAG

Adds `ParallelWindowRule` to the AQE default_optimizers chain (position 2,
before `SelectJoinRule` and DF's own optimizers, before `DistributedExchangeRule`).
The rule matches bounded RANGE-frame windows with no PARTITION BY and a
single-column Float64 ORDER BY, and rewrites them into a range-shuffle
so BoundedWindowAggExec's SinglePartition requirement isn't a serial
bottleneck. Shape:

  RangeFilterExec (narrow, halo=0, cuts=pending)
    BoundedWindowAggExec
      SortPreservingMergeExec
        RangeFilterExec (wide, halo=frame bounds, cuts=pending)
          RuntimeStatsExec (post-ORRE per-partition sketch → scheduler)
            OrderedRangeRepartitionExec (K sorted disjoint outputs)
              RuntimeStatsExec (local sketch; feeds ORRE's cut walker)
                SortExec (preserve_partitioning=true)
                  <source>

Both RangeFilterExecs are planted with cuts=None. After stage 0's tasks
complete and their RSE reports are merged into K-1 quantile cuts, the
scheduler's `resolve_range_filter_cuts` walker (in `adapt_to_ballista`)
finds every pending RangeFilterExec in the downstream stage's plan and
resolves it against the matching ExchangeExec's routing_expr. Adapter no
longer injects RangeFilterExec — the rule is the sole planter, single
source of truth. Idempotency guard on the rule bails when the BWAG's
subtree already contains our own ORRE/RangeFilter (AQE re-plans fire the
chain again on the already-rewritten plan).

Also relaxes `ORRE::try_new` — the child-claims-sortedness check moved
from construction to `execute()`. Rule-time construction races with
`EnforceSorting` (which planted a SortExec on ORRE's declared
`required_input_ordering` *after* the rule ran), so refusing at try_new
was too strict. The runtime check at execute() still catches invariant
breaks; two tests moved from `try_new_rejects_*` to `execute_rejects_*`.

Verified on h2o Q8 at 1e7 under a 2G/exec cgroup cap: stage 0 (8 tasks)
and stage 1 (8 tasks) both parallelize across both executors, no OOM.
Stage 2 still collapses to a single task doing SPM+BWAG+narrow — DE
inserts a shuffle boundary below the SPM, putting BWAG in the final
stage. That collapse is the next follow-up; the machinery for the
range-shuffle itself is in.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

fix(scheduler): stop DE inserting Exchange between SPM and rule-planted RangeFilterExec

`ParallelWindowRule` plants a `RangeFilterExec` directly on the resolved
range-repartition `ExchangeExec`, with `SortPreservingMergeExec` above.
`DistributedExchangeRule`'s SPM branch was checking whether SPM's
immediate child was an `ExchangeExec` — seeing the `RangeFilterExec`, it
injected another `ExchangeExec`, cutting the plan into an extra collapse
stage.

Introduce `is_stage_boundary` and treat a `RangeFilterExec` sitting
directly on an `ExchangeExec` as part of the boundary. That is a
conscious design shape — we chose not to fold range-filtering into
`ShuffleReader`/`ExchangeExec`, so the filter is part of the boundary
by construction. This matches the pre-`fcb31520` behaviour where the
adapter injected the filter after DE had already run.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

refactor(scheduler): size ParallelWindowRule's K from source partitions

K was `config.execution.target_partitions.max(2)` — a placeholder
chosen while writing the rule. The natural sizing is
`source.output_partitioning().partition_count()`: ORRE re-slices each
input partition into a range-disjoint output partition, so K = input
partitions is the 1:1 rearrangement.

No behaviour change on h2o Q8 (`target_partitions` and source
partitions both settle at 8), but the rule no longer depends on the
config knob or its `.max(2)` fallback.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

feat(core, scheduler): PartitionedBoundedWindowAggExec — parallel BWAG for the range-window shape

DataFusion's `BoundedWindowAggExec` declares `SinglePartition` when no
PARTITION BY is present, forcing `EnforceDistribution` to collapse K→1
via `SortPreservingMergeExec`. With `ParallelWindowRule`'s range-
repartition upstream, each ORRE output partition is a globally
range-disjoint slice + halo — BWAG can safely run per-partition on those
K slices and produce K correct outputs. `PartitionedBoundedWindowAggExec`
wraps BWAG, exposes only the input as its plan-tree child (BWAG itself
is hidden from tree walkers), and overrides `required_input_distribution`
to `UnspecifiedDistribution`. `execute(i)` delegates to the wrapped BWAG,
which already processes each partition independently.

- `ballista_core::execution_plans::partitioned_bounded_window_agg`: the
  new operator. `InputOrderMode` and `can_repartition` are hardcoded
  (`Sorted` / `false`) per the rule's shape gates.
- `BallistaPhysicalPlanNode::PartitionedBoundedWindowAgg`: proto message
  carrying only `window_expr` — the rest is implicit from the rule's
  invariants. Round-trip goes through DF's
  `serialize_physical_window_expr` / `parse_physical_window_expr`.
- `ParallelWindowRule::rewrite_bwag`: drops the SPM the previous
  rewrite planted between BWAG and the wide `RangeFilterExec`, and
  swaps BWAG for `PartitionedBoundedWindowAggExec`. K is sourced from
  `config.execution.target_partitions.max(2)` — at rule-fire time
  `DataSourceExec` still has 1 file_group (splits happen later in the
  AQE chain), so the plan tree can't yet tell us the true source
  width. Reverts the "size K from source" refactor.
- `rewrites_q8_shape` test now asserts `PartitionedBoundedWindowAggExec`
  and NO `SortPreservingMergeExec` in the output.

On h2o Q8 @ 1e7 under a 2G/exec cgroup cap, 2 execs × 4 vcores,
`ballista.scheduler.max_partitions_per_task=4`: 41 s (down from 155 s)
and returns the full 10M rows (previous runs returned only 1.55M —
the K→1 collapse dropped ~87% of the output because the narrow
`RangeFilterExec` above the collapsed BWAG kept only partition-0's
range). Both stages run 2 MPT tasks (one per exec).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

feat(core, scheduler): gate ParallelWindowRule behind ballista.planner.parallel_window.enabled

Adds an opt-in config flag so users of AQE don't inherit the range-window
rewrite by default. Matches the shape of `ballista.planner.coalesce.enabled`:
new AQE rule → new opt-in flag. Default `false`.

- `BALLISTA_PARALLEL_WINDOW_ENABLED` + registry entry + getter on
  `BallistaConfig`.
- Guard clause at the top of `ParallelWindowRule::optimize` returns the
  plan untouched when the flag is off.
- Existing shape tests keep the rule enabled through the local `optimize`
  helper; a new `disabled_by_default` test asserts the rewrite is inert
  without the extension registered.
- Regenerated `docs/source/user-guide/configs.md`.

notes

feat(core, scheduler, executor): RangeShuffleReaderExec — ordered k-way merge at stage boundary [WIP]

Closes the RANGE-frame correctness gap: the regular ShuffleReaderExec
concatenates upstream sources in arrival order, breaking the monotonicity
BWAG's Range-frame cursor assumes. RangeShuffleReaderExec keeps each source
alive as its own stream and feeds them all into StreamingMerge on the
child's declared ordering.

- new RangeShuffleReaderExec (fetch reuses shuffle_reader helpers, backpressure
  via merge demand; no permit governor, no per-source buffering)
- adapter plants it whenever exchange.input().output_ordering().is_some()
- proto + codec round-trip; executor work_dir/client_pool late binding;
  task_builder partition-slice restriction

h2o Q8 @ 1e7 SUM diff: parallel_window=true/false now agree to 5e-14 relative
(FP noise floor). Previously diverged at run boundaries.

Follow-ups (see next-session TODO): planner.rs::rollback_resolved_shuffles
falls through, cluster/mod.rs::stage_has_input_collapse falls through.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

refactor(scheduler): audit-driven RangeShuffleReaderExec whitelist fills

Fills two production downcast sites that fell through the RangeShuffleReaderExec
shape, plus fmt fallout from the initial slice.

- planner::rollback_resolved_shuffles: rolls range readers back to plain
  UnresolvedShuffleExec. Range-ness is derived at plan time from the child's
  ordering, so a re-plan's adapter walk re-plants a fresh range reader — no
  proto extension needed.
- cluster::stage_has_input_collapse: range reader is a stage boundary; the
  walker must stop there, else a single-output-partition range reader
  spuriously trips the `partition_count == 1` collapse arm.

Tests:
- rollback_resolved_shuffles_reduces_range_reader_to_plain_unresolved
- stage_has_input_collapse_stops_at_range_reader

Fmt: adapter's #[cfg(test)] mod moved below resolve_range_filter_cuts to
satisfy clippy::items_after_test_module.

Follow-up still open: execution_graph_dot.rs graphviz — will render generic
node label for the range reader. Diagnostic only, safe to punt.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

docs(parallel-range-window): tick Ordered ShuffleReader, drop correctness-gap section

The gap is closed — RangeShuffleReaderExec ships in d87321d + c8f3f20, and
the h2o Q8 SUM diff between parallel_window=true/false lands at 5e-14 relative
(Float64 noise floor). Rewrite the ticked line to describe the landed shape
and note the writer-vs-demand-driven follow-up. Drop the "Dot-product check"
bullet for the reader (now landed) and the whole correctness-gap section.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

perf(core): RangeFilterExec min/max fast paths + binary-search slice

When `input.output_ordering()` leads with `routing_expr` ascending, take one
of three shortcuts on each batch before `filter_record_batch`:

  - `last < lo` or `first >= hi` → drop the whole batch (skip).
  - `first >= lo && last < hi`   → pass the batch through unchanged (Arc-clone).
  - mixed                         → `partition_point` on the Float64Array values
                                    for lo/hi indices + `RecordBatch::slice`
                                    (zero-copy view).

Nullable routing columns fall back to `filter_record_batch` on a per-batch
basis (Float64Array::values() returns garbage for null slots, breaking
partition_point). `sorted_on_key` is derived at construction — no config knob.

h2o Q8 with 2 execs × 4 vcores × MPT=4:

  scale  cap  parallel_window=false  parallel_window=true  speedup
  1e7    2G   7.6 s                  2.5 s                 3.0×
  1e8    4G   143 s                  92 s                  1.55×

The 1e8 delta is smaller because the bottleneck shifts to shuffle IO /
whole-file merge memory — the ValueIndex + per-task halo work next.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

fix(docs): rustdoc + prettier CI

- rustdoc: three `[`resolve_cuts`]` references in `range_filter.rs` (module
  doc + two item docs) resolved to no target; qualify as
  `RangeFilterExec::resolve_cuts` / `Self::resolve_cuts` so cargo doc no
  longer errors on ballista-core.
- prettier: `docs/developer/parallel-range-window.md` had two
  `*emphasis*` spans (`*shape*`, `*task-level*`) and a stray blank line —
  prettier wants `_emphasis_` + single blank. No content change.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

feat(core): RangeFilterExec metrics — fast-path counters + baseline

Was returning `None` from `metrics()`, so the operator was invisible in
the scheduler's stage-metrics dump. Adds `ExecutionPlanMetricsSet` with
`BaselineMetrics` (elapsed_compute, output_rows via record_poll) and
five path counters: `fast_skip_batches`, `fast_pass_batches`,
`fast_slice_batches`, `slow_batches`, `input_rows`. Timer scoped
post-poll so upstream shuffle IO isn't billed to this op.

Scout on h2o Q8 @ 1e8 confirms fast path is firing as intended
(99%+ pass-through on the narrow filter, 85% skip on the wide one,
zero slow-path fallbacks) — filter is not the perf bottleneck.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

fix(docs): drop unresolved intra-doc link in parallel_window

`resolve_range_filter_cuts` is private to the adapter module and not
in scope from `parallel_window.rs`, so rustdoc rejects the intra-doc
link under `-D warnings`. Keep it as plain inline code.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

style(core): join split struct decl to satisfy rustfmt

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

refactor(core): slim RangeFilterExec — scheduler owns cuts/partition indices, RFE widens by halo

Split the range-partitioning concerns out of RangeFilterExec so it looks
like PerPartitionFilterExec's counterpart for sorted-key filtering:

- RFE fields: input, routing_expr, halo_lo/hi (ScalarValue), raw_bounds
  (late-bound), sorted_on_key detection, metrics.
- Gone: cuts, partition_indices, resolve_cuts, restrict_partitions,
  try_new_with_indices. Widening from cuts+halo to per-partition bounds
  moves scheduler-side (adapter builds raw_bounds from cuts; RFE widens
  by its own halos internally at resolve_bounds time).
- task_builder RFE branch is now a plain "slice raw_bounds parallel to
  input restriction" — no partition_indices remap.
- All APIs and proto fields are ScalarValue (arrow-primitive-generic);
  internal downcast to f64 with Err for non-Float64 until KLL widens.

Halos are functional on RFE (widens raw→widened at resolve time), not
write-only decoration. The scheduler-side cut_partitions also needs
halo-widened overlap for correct file routing to RANGE-frame consumers;
that's a separate cross-stage lookup left as a TODO here.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

docs(core): fix RangeFilterExec intra-doc link at module scope

`[\`Self::resolve_bounds\`]` on line 39 was in the module-level `//!`
comment where `Self` is not defined. CI runs cargo doc with -D warnings
so it fails; local runs pass silently. Use the fully-qualified path.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

docs(scheduler): revert PPFE→RFE comment renames to reduce PR diff

Three files carried only doc/comment renames from PerPartitionFilterExec to
RangeFilterExec — no code changes. PPFE still exists in the tree, so the
original phrasing remains accurate. Reverting shrinks the PR's review
surface without touching semantics; a follow-up sweep can update these
comments after PPFE is fully retired.

- exchange.rs: 3 comment mentions of PPFE-as-cuts-consumer
- test/coalesce_rule.rs: 1 test comment
- test/range_repartition.rs: 2 test comments

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

fix(scheduler): plant RSE#1 below SortExec so cuts land before ORRE routes

`ParallelWindowRule` used to leave `RSE#1` above `SortExec`, so the local
sketch only started ingesting after Sort had fully materialized. ORRE
consumed from RSE#1 as soon as Sort emitted, meaning the scheduler often
handed ORRE a still-being-built sketch → approximate cuts → skewed
shuffle files.

Two changes to close this:

1. Move the rule to run *after* the DataFusion optimizer chain. At the
   old position the input was `BWAG → DataSource` (sources with
   `sort_order_for_reorder` satisfy BWAG's ordering natively, no Sort
   inserted yet); the SortExec placement we care about is only
   materialized once EnforceSorting / RepartitionFileScans have run.
   Running earlier also lets DF's later sort-pushdown move any Sort we
   plant down through the passthrough RSE#1, undoing the intended order.

2. Strip whatever DF planted for BWAG's SinglePartition + Sorted
   requirements (SPM and/or SortExec) and plant a fresh
   `SortExec → RSE#1 → source` chain below `ORRE`. The fresh Sort is
   the pipeline break: it consumes all input before emitting the first
   row, so RSE#1's sketch fully reports while Sort buffers.

Q8 (h2o, SF=1e7, 8 vcores):
- rule skipped (buggy pattern):  61s
- RSE#1 above Sort (prior):      24s
- RSE#1 below Sort (this):       17s  ← ~1.4× speedup over prior

Wide-RFE metrics on the new plan: input=18.84M / output=11.01M against a
10M-row dataset → 1.88× row-level read amplification, matching the
theoretical (1 + halo/cut_width) ≈ 1.24× floor plus batch-granularity
overhead from RangeShuffleReader.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

� Conflicts:
�	ballista/core/proto/ballista.proto
�	ballista/core/src/execution_plans/mod.rs
�	ballista/core/src/execution_plans/range_filter.rs
�	ballista/core/src/serde/generated/ballista.rs